//
// Created by LiuYou on 2021/8/19.
//

#ifndef HIGHPERFORMANCELINUXSERVERPROGRAMMING_LINUX2_PROCESSPOOL_HPP
#define HIGHPERFORMANCELINUXSERVERPROGRAMMING_LINUX2_PROCESSPOOL_HPP


#include <unistd.h>
#include <memory>
#include <sys/epoll.h>
#include <cassert>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/wait.h>


/** 处于处理信号的管道, 以实现统一事件源。后面称之为信号管道。*/
static int signalPipeFd[2];


extern int setFdNonblocking(int fd);

extern void addFdToEpoll(int epollFd, int fd);

extern void addSignal(int sig, void(* signalHandler)(int), bool restart = true);

extern void signalHandler(int sig);

/**
 * @brief 从epollFd标识的epoll内核事件表中删除fd上的所有注册事件。
 * @param epollFd
 * @param fd
 */
extern void removeFd(int epollFd, int fd);


/**
 * @brief 描述一个子进程的类。
 * pid是目标子进程的PID, pipeFd是父子进程间通信用的管道。
 */
class Process {
public:
    Process();

    ~Process();

public:
    pid_t pid;
    int pipeFd[2];
};

/**
 * @brief 进程池类, 将它定义为模板类是为了代码复用。其模板参数是处理逻辑任务的类。
 */
template<typename T>
class ProcessPool {
public:
    /** 启动进程池。
     * <p>
     * 父进程中的idx值为-1, 子进程中idx值大于等于0, 我们据此判断接下来要运行的是父进程代码还是子进程代码。*/
    void run() {
        if (idx != -1) {
            runChild();
            return;
        }
        runParent();
    }

private:
    /** 将构造函数定义为私有, 因此我们只能通过后面的create()静态函数来创建ProcessPool实例。*/
    explicit ProcessPool(int listenFd) :
            listenFd( listenFd ), idx( -1 ), stop( false ), epollFd() {
        assert((processNumber > 0) && (processNumber <= MAX_PROCESS_NUMBER));

        for (int i = 0; i < processNumber; ++i) {
            subProcess[i] = std::make_unique<Process>();
        }

        /* 创建processNumber个子进程, 并建立它们和父进程之间的管道。*/
        for (int i = 0; i < processNumber; ++i) {
            int ret = socketpair( PF_UNIX, SOCK_STREAM, 0, subProcess[i]->pipeFd );
            assert( ret == 0 );

            subProcess[i]->pid = fork();
            assert( subProcess[i]->pid >= 0 );
            if (subProcess[i]->pid > 0) {
                close( subProcess[i]->pipeFd[1] );
                continue;
            } else {
                close( subProcess[i]->pipeFd[0] );
                idx = i;
                break;
            }
        }
    }


private:
    /** 统一事件源。*/
    void setupSignalPipe() {
        /* 创建epoll事件监听表和信号管道。*/
        epollFd = epoll_create( 5 );
        assert( epollFd != -1 );

        socketpair( PF_UNIX, SOCK_STREAM, 0, signalPipeFd );

        setFdNonblocking( signalPipeFd[1] );
        addFdToEpoll( epollFd, signalPipeFd[0] );

        /* 设置信号处理函数。*/
        addSignal( SIGCHLD, signalHandler );
        addSignal( SIGTERM, signalHandler );
        addSignal( SIGINT, signalHandler );
        addSignal( SIGPIPE, SIG_IGN);
    }

    void runParent() {
        setupSignalPipe();

        /* 父进程监听listenFd。*/
        addFdToEpoll( epollFd, listenFd );

        epoll_event epollEvents[MAX_EVENT_NUMBER];
        int subProcessCounter = 0;
        int newConnect = 1;

        while (!stop) {
            int readyFdNum = epoll_wait( epollFd, epollEvents, MAX_EVENT_NUMBER, -1 );
            if ((readyFdNum < 0) && (errno != EINTR)) {
                printf( "epoll failure\n" );
                break;
            }

            for (int i = 0; i < readyFdNum; ++i) {
                int fd = epollEvents[i].data.fd;
                if (fd == listenFd) {
                    /* 如果有新连接到来, 就采用 Round Robin 方式将其分配给一个子进程处理。*/
                    int childPid = subProcessCounter;
                    do {
                        if (subProcess[childPid]->pid == -1) {
                            break;
                        }
                        childPid = (childPid + 1) % processNumber;
                    } while (childPid != subProcessCounter);

                    if (subProcess[childPid]->pid == -1) {
                        stop = true;
                        break;
                    }
                    subProcessCounter = (childPid + 1) % processNumber;
                    send( subProcess[childPid]->pipeFd[0], (char*) &newConnect, sizeof( newConnect ), 0 );
//                    printf( "send request to child %d\n", childPid );
                    printf( "向子进程: %d, 通过管道, 发送一个请求。\n", childPid );
                    fflush( stdout );
                }
                    /* 下面处理父进程接收到的信号。*/
                else if ((fd == signalPipeFd[0]) && (epollEvents[i].events & EPOLLIN)) {
                    char signals[1024];
                    ssize_t rcvByteNum = recv( signalPipeFd[0], signals, sizeof( signals ), 0 );
                    if (rcvByteNum <= 0) {
                        continue;
                    } else {
                        for (int j = 0; j < rcvByteNum; ++j) {
                            switch (signals[j]) {
                                case SIGCHLD: {
                                    pid_t pid;
                                    int stat;
                                    while ((pid = waitpid( -1, &stat, WNOHANG )) > 0) {
                                        for (int k = 0; k < processNumber; ++k) {
                                            /* 如果进程池中第i个子进程退出了, 则主进程关闭相应的通信管道, 并设置
                                             * 相应的pid为-1, 以标记该子进程已经退出。*/
                                            if (subProcess[k]->pid == pid) {
                                                printf( "child %d join\n", k );
                                                close( subProcess[k]->pipeFd[0] );
                                                subProcess[k]->pid = -1;
                                            }
                                        }
                                    }
                                    /* 如果所有子进程都已经退出了, 则父进程也退出。*/
                                    stop = true;
                                    for (int k = 0; k < processNumber; ++k) {
                                        if (subProcess[k]->pid != -1) {
                                            stop = false;
                                        }
                                    }
                                    break;
                                }
                                case SIGTERM: {
                                    break;
                                }
                                case SIGINT: {
                                    /* 如果父进程接受到终止信号, 那么就杀死所有子进程, 并等待它们全部结束。当然, 通知
                                     * 子进程结束更好的方法是向父子进程间的通信管道发送特殊数据, 读者不妨自己实现之。*/
                                    printf( "kill all the child now\n" );
                                    for (int k = 0; k < processNumber; ++k) {
                                        int pid = subProcess[k]->pid;
                                        if (pid != -1) {
                                            kill( pid, SIGTERM );
                                        }
                                    }
                                    break;
                                }
                                default: {
                                    break;
                                }
                            }
                        }
                    }
                } else {
                    continue;
                }
            }
        }

        close( epollFd );
    }

    void runChild() {
        setupSignalPipe();

        /* 每个子进程都通过其在进程池中的序号值idx找到与父进程通信的管道。*/
        int pipeFd = subProcess[idx]->pipeFd[1];
        /* 子进程需要监听管道文件描述符pipeFd, 因为父进程将通过它来通知子进程accept新连接。*/
        addFdToEpoll( epollFd, pipeFd );

        epoll_event epollEvents[MAX_EVENT_NUMBER];
        std::array<std::unique_ptr<T>, USER_PER_PROCESS> users;
        for (int i = 0; i < USER_PER_PROCESS; ++i) {
            users[i] = std::make_unique<T>();
        }

        while (!stop) {
            int readyFdNum = epoll_wait( epollFd, epollEvents, MAX_EVENT_NUMBER, -1 );
            if ((readyFdNum < 0) && (errno != EINTR)) {
                printf( "epoll failure\n" );
                break;
            }

            for (int i = 0; i < readyFdNum; ++i) {
                int fd = epollEvents[i].data.fd;
                if ((fd == pipeFd) && epollEvents[i].events & EPOLLIN) {
                    int client = 0;
                    /* 从父子进程间的管道读取数据, 并将结果保存在变量client中。如果读取成功, 则表示有辛苦胡连接到来。*/
                    ssize_t rcvByteNum = recv( fd, (char*) &client, sizeof( client ), 0 );
                    if (((rcvByteNum < 0) && (errno != EAGAIN)) || rcvByteNum == 0) {
                        continue;
                    } else {
                        sockaddr_in clientAddress{};
                        socklen_t socklen = sizeof( clientAddress );
                        int connectFd = accept( listenFd, (sockaddr*) &clientAddress, &socklen );
                        if (connectFd < 0) {
                            perror( "accept()" );
                            continue;
                        } else {
                            addFdToEpoll( epollFd, connectFd );
                            /* 模板类T必须实现init()方法, 以初始化一个客户连接。我们直接使用connectFd来索引逻辑处理对象
                             * (T类型的对象), 以提高程序效率。*/
                            users[connectFd]->init( connectFd, <#initializer#>, <#initializer#>, <#initializer#>,
                                                    clientAddress, epollFd );
                        }
                    }
                } else if ((fd == signalPipeFd[0]) && (epollEvents[i].events & EPOLLIN)) {
                    char signals[1024];
                    ssize_t rcvByteNum = recv( signalPipeFd[0], signals, sizeof( signals ), 0 );
                    if (readyFdNum <= 0) {
                        continue;
                    } else {
                        for (int j = 0; j < rcvByteNum; ++j) {
                            switch (signals[j]) {
                                case SIGCHLD: {
                                    pid_t pid;
                                    int stat;
                                    while ((pid = waitpid( -1, &stat, WNOHANG )) > 0) {
                                        continue;
                                    }
                                    break;
                                }
                                case SIGTERM: {
                                    break;
                                }
                                case SIGINT: {
                                    stop = true;
                                    break;
                                }
                                default: {
                                    break;
                                }
                            }
                        }
                    }
                }
                    /* 如果是其它可读数据, 那么必然是客户请求到来。调用逻辑处理对象的process方法处理之。*/
                else if (epollEvents[i].events & EPOLLIN) {
                    users[fd]->process();
                } else {
                    continue;
                }
            }
        }
        close( pipeFd );
        close( epollFd );
    }

public:
    /** 单例模式, 以保证程序最多创建一个ProcessPool实例, 这是程序正确处理信号的必要条件。*/
//    static std::unique_ptr<ProcessPool<T>> getInstance(int listenFd);
    static ProcessPool<T>* getInstance(int listenFd) {
        if (instance == nullptr) {
            instance = new ProcessPool<T>( listenFd );
        }
        return instance;
    }

    ~ProcessPool() = default;

private:
    /** 进程池允许的最大子进程数量。*/
    static const int MAX_PROCESS_NUMBER = 16;

    /** 每个子进程最多能处理的客户数量。*/
    static const int USER_PER_PROCESS = 65536;

    /** epoll最多能处理的事件数。*/
    static const int MAX_EVENT_NUMBER = 10000;

    /** 进程池中的进程总数。*/
    static const int processNumber = 8;

    /** 子进程在池中的序号, 从0开始。*/
    int idx;

    /** 每个进程都有一个epoll内核事件表, 用 epollFd标识。*/
    int epollFd;

    /** 监听socket。*/
    int listenFd;

    /** 子进程通过 stop来决定是否停止运行。*/
    bool stop;

    /** 保存所有子进程的描述信息。*/
//    std::unique_ptr<Process> subProcess;
    std::array<std::unique_ptr<Process>, processNumber> subProcess;

    /** 进程池静态实例。*/
    // 如果使用 unique_ptr来实现单例模式。
//    /*static*/ std::unique_ptr<ProcessPool<T>> instance;

    // 放弃了使用 unique_ptr来实现单例模式。
    static ProcessPool<T>* instance;
};

template<typename T>
ProcessPool<T>* ProcessPool<T>::instance = nullptr;


///////////////////////////////////////////////////////////////////////////

//template<typename T>
//std::unique_ptr<ProcessPool<T>>
//ProcessPool<T>::getInstance(int listenFd) {
///*    if (!instance.get()) {
//        instance = std::make_unique<ProcessPool<T>>( listenFd );
//    }
//    return instance;*/
//
//
//    return std::make_unique<ProcessPool<T>>( listenFd );
//}




#endif //HIGHPERFORMANCELINUXSERVERPROGRAMMING_LINUX2_PROCESSPOOL_HPP
